package com.bingxu.flink.action;

import com.bingxu.flink.bean.UserBehavior;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import scala.Int;

import java.util.HashSet;
import java.util.Set;

/**
 * @author :BingXu
 * @description :TODO
 * @date :2021/8/13 8:24
 * @modifier :
 */

public class UV_Count1 {
    public static void main(String[] args) throws Exception {
        // 运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // load data and transform Java Bean object
        DataStreamSource<String> sourceDS = env.readTextFile("D:\\Source\\JavaProject\\flinkdemo\\input\\UserBehavior.csv");
        SingleOutputStreamOperator<UserBehavior> userBehaviorDS = sourceDS.map(new MapFunction<String, UserBehavior>() {
            @Override
            public UserBehavior map(String s) throws Exception {
                String[] fields = s.split(",");
                return new UserBehavior(Long.parseLong(fields[0]),
                        Long.parseLong(fields[1]),
                        Integer.parseInt(fields[2]),
                        fields[3],
                        Long.parseLong(fields[4])
                );
            }
        });
        // 通过累加器进行处理
        SingleOutputStreamOperator<UserBehavior> filterDS = userBehaviorDS.filter(ele -> "pv".equals(ele.getBehavior()));
        // 1.对于用户id进行分组，分组之后 userid
        filterDS
                .map((MapFunction<UserBehavior, Long>) value -> value.getUserId())
                .setParallelism(1)
                .keyBy(ele -> ele)
                .process(new ProcessFunction<Long, Integer>() {
                    private int sum=0;
                    @Override
                    public void processElement(Long value, Context ctx, Collector<Integer> out) throws Exception {
                        System.out.println();
                        out.collect(sum++);
                    }
                }).setParallelism(1)
                .print().setParallelism(1);

        // 2.利用set集合去重
//        filterDS.map(new MapFunction<UserBehavior, Long>() {
//
//            @Override
//            public Long map(UserBehavior value) throws Exception {
//                return value.getUserId();
//            }
//        })
//                .process(new ProcessFunction<Long, Integer>() {
//                    private Set<Long> set = new HashSet<>();
//
//                    @Override
//                    public void processElement(Long value, Context ctx, Collector<Integer> out) throws Exception {
//                        set.add(value);
//                        out.collect(set.size());
//                    }
//                })
//                .setParallelism(1)
//                .print()
//                .setParallelism(1);

        env.execute();

    }
}
